package com.dhz.common.monitor;

import com.alibaba.fastjson.JSON;
import com.dhz.common.utils.NetUtils;
import com.dhz.common.utils.http.HttpClientUtils;
import com.google.common.primitives.Longs;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 监控处理类
 *
 * @author FY
 * @since 1.0
 */
@Aspect
public class MonitorAspect implements MonitorService {
    private static final Logger logger = LoggerFactory.getLogger(MonitorAspect.class);

    // --------------------------------------------------------------------
    /** 监控数据数量 */
    private static final int LENGTH = 6;
    // 定时任务执行器
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3, new NamedThreadFactory("MonitorSendTimer", true));
    // 定时器执行间隔,默认
    private static final long MONITOR_INTERVAL_DEFAULT = 60000; // ms

    // 定时器执行间隔
    private long monitorInterval; // ms
    // 监控数据存储服务器地址
    private String monitorServerAddr;

    private String application;
    private String version;

    private final ConcurrentMap<String, AtomicInteger> concurrents = new ConcurrentHashMap<String, AtomicInteger>();
    private final ConcurrentMap<Statistics, AtomicReference<long[]>> statisticsMap = new ConcurrentHashMap<Statistics, AtomicReference<long[]>>();

    // 统计信息收集定时器
    private final ScheduledFuture<?> sendFuture;

    public MonitorAspect(String monitorServerAddr, String application, String version) {
        this(monitorServerAddr, application, version, MONITOR_INTERVAL_DEFAULT);
    }

    public MonitorAspect(String monitorServerAddr, String application, String version, long monitorInterval) {
        if (monitorServerAddr == null || monitorServerAddr.length() == 0) {
            throw new IllegalArgumentException("监控数据的存储服务器传输地址不能为空");
        }
        if (application == null || application.length() == 0) {
            throw new IllegalArgumentException("监控的服务名称不能为空");
        }
        if (monitorInterval < 1000) {
            monitorInterval = 1000;
        }

        this.monitorServerAddr = monitorServerAddr;
        this.monitorInterval = monitorInterval;
        this.version = version;
        this.application = application;

        sendFuture = scheduledExecutorService.scheduleWithFixedDelay(this::send, this.monitorInterval, this.monitorInterval, TimeUnit.MILLISECONDS);
    }
    // --------------------------------------------------------------------

    /**
     * 切入点定义，带有{@link MonitorAble} 注解的所有方法
     */
    @Pointcut("execution(@MonitorAble public void *(..))")
    private void pointCutMethodNoResult() {}

    /**
     * 切入点定义，带有{@link MonitorAble} 注解的所有方法
     */
    @Pointcut("execution(@MonitorAble public * *(..))")
    private void pointCutMethodResult() {}

    /**
     * 对{@link #pointCutMethodNoResult()}方法定义的切入点配置环绕通知
     * @param pjp {@link ProceedingJoinPoint} 连接点对象
     */
    @Around(value = "pointCutMethodNoResult()")
    public void aroundNoResult(ProceedingJoinPoint pjp) throws Throwable {
        // 监控对象数据收集
        String service = pjp.getSignature().getDeclaringTypeName();
        String method = pjp.getSignature().getName();
        String client = "";
        long start = System.currentTimeMillis();
        try {
            getConcurrent(service, method).incrementAndGet(); // 并发计数
            // 执行方法
            pjp.proceed();
            collect(service, method, client, 1, 0, start);
        } catch (Throwable throwable) {
            collect(service, method, client, 0, 1, start);
            throw throwable;
        }
    }

    /**
     * 对{@link #pointCutMethodResult()}方法定义的切入点配置环绕通知
     * @param pjp {@link ProceedingJoinPoint} 连接点对象
     */
    @Around(value = "pointCutMethodResult()")
    public Object aroundResult(ProceedingJoinPoint pjp) throws Throwable {

        Object result;

        // 监控对象数据收集
        String service = pjp.getSignature().getDeclaringTypeName();
        String method = pjp.getSignature().getName();
        String client = "";
        long start = System.currentTimeMillis();

        try {
            getConcurrent(service, method).incrementAndGet(); // 并发计数
            // 执行方法
            result = pjp.proceed();
            collect(service, method, client, 1, 0, start);
        } catch (Throwable throwable) {
            collect(service, method, client, 0, 1, start);
            throw throwable;
        }
        return result;
    }

    // 收集监控数据
    public void collect(String service, String method, String client, long success, long failure, long start) {
        int concurrent = getConcurrent(service, method).get(); // 当前并发数
        // 初始化原子引用
        Statistics statistics = new Statistics(application, service, method, null, version, client, NetUtils.getLocalHost());
        AtomicReference<long[]> reference = statisticsMap.get(statistics);
        if (reference == null) {
            statisticsMap.putIfAbsent(statistics, new AtomicReference<long[]>());
            reference = statisticsMap.get(statistics);
        }

        // 耗时
        long elapsed = System.currentTimeMillis() - start;

        // CompareAndSet并发加入统计数据
        long[] current;
        long[] update = new long[LENGTH];

        do {
            current = reference.get();
            if (current == null) {
                update[0] = success;
                update[1] = failure;
                update[2] = elapsed;
                update[3] = concurrent;
                update[4] = elapsed;
                update[5] = concurrent;
            } else {
                update[0] = current[0] + success;
                update[1] = current[1] + failure;
                update[2] = current[2] + elapsed;
                update[3] = (current[3] + concurrent) / 2;
                update[4] = current[4] > elapsed ? current[4] : elapsed;
                update[5] = current[5] > concurrent ? current[5] : concurrent;
            }
        } while (!reference.compareAndSet(current, update));
    }

    /**
     * 发送监控数据到Monitor数据存储服务器
     */
    public void send() {
        for (Map.Entry<Statistics, AtomicReference<long[]>> entry : statisticsMap.entrySet()) {
            // 获取已统计数据
            Statistics statistics = entry.getKey();
            AtomicReference<long[]> reference = entry.getValue();

            long[] numbers = reference.get();
            long success = numbers[0];
            long failure = numbers[1];
            long elapsed = numbers[2];
            long concurrent = numbers[3];
            long maxElapsed = numbers[4];
            long maxConcurrent = numbers[5];

            //

            // 减掉已统计数据
            long[] current;
            long[] update = new long[LENGTH];
            do {
                current = reference.get();
                if (current == null) {
                    update[0] = 0;
                    update[1] = 0;
                    update[2] = 0;
                    update[3] = 0;
                } else {
                    update[0] = current[0] - success;
                    update[1] = current[1] - failure;
                    update[2] = current[2] - elapsed;
                    update[3] = current[3] - concurrent;
                }
            } while (! reference.compareAndSet(current, update));

            // 发送数据到服务器
            postMonitorData(statistics, current);

        }
    }

    /**
     * 提交监控数据到监控数据存储服务器
     * @param statistics {@link Statistics}
     * @param current long[]
     */
    private void postMonitorData(final Statistics statistics, long[] current) {
        long timestamp = System.currentTimeMillis();
        Map<String, String> queryParams = new HashMap<String, String>();
        Map<String, String> formParams = new HashMap<String, String>();

        formParams.put("statistics", JSON.toJSONString(statistics));
        formParams.put("data", Longs.join(",", current));
        formParams.put("timestamp", String.valueOf(timestamp));

        try {
            HttpClientUtils.getInstance().doPostForString(monitorServerAddr, queryParams, formParams);
//            logger.debug("发送监控数据到GMS服务器[{}]成功，耗时 : {} ms.", this.monitorServerAddr, (System.currentTimeMillis() - timestamp));
        } catch (URISyntaxException | IOException e) {
            logger.warn("发送监控数据到GMS服务器[{}]失败，耗时 : {} ms.", this.monitorServerAddr, (System.currentTimeMillis() - timestamp)/*, e*/);
        }
    }

    /**
     * 收集数据
     * @param statistics {@link Statistics}
     */
    @Override
    public void collect(Statistics statistics) {
        // do something...
    }

    @Override
    public void destroy() {
        try {
            sendFuture.cancel(true);
            scheduledExecutorService.shutdown();
        } catch (Throwable t) {
            logger.warn("关闭定时器和线程池失败 : ", t);
        }
    }

    // 获取并发计数器
    private AtomicInteger getConcurrent(String service, String method) {
        // 获得请求对应的Controller#method名称
        String key = service + "#" + method;
        AtomicInteger concurrent = concurrents.get(key);
        if (concurrent == null) {
            concurrents.putIfAbsent(key, new AtomicInteger());
            concurrent = concurrents.get(key);
        }
        return concurrent;
    }
}
